package cn.gwm.flink.streaming.function.process;
import cn.gwm.flink.streaming.beans.termalRunaway.BmsTrdStrategys;
import cn.gwm.flink.streaming.beans.termalRunaway.WaringTermalDwmWithStragegyEv;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * @author GW00283474
 */
public class TermalWarningEvBroadcastProcessFunction extends KeyedBroadcastProcessFunction<String, WaringTermalDwmWithStragegyEv, Map<String, List>, WaringTermalDwmWithStragegyEv> {

    private MapStateDescriptor<String, List> mapStateDescriptor;

    public TermalWarningEvBroadcastProcessFunction(MapStateDescriptor<String, List> MAP_STATE_DESCRIPTOR) {
        this.mapStateDescriptor = MAP_STATE_DESCRIPTOR;
    }

    @Override
    public void processElement(WaringTermalDwmWithStragegyEv record, KeyedBroadcastProcessFunction<String, WaringTermalDwmWithStragegyEv, Map<String, List>, WaringTermalDwmWithStragegyEv>.ReadOnlyContext ctx, Collector<WaringTermalDwmWithStragegyEv> out) throws Exception {
        ReadOnlyBroadcastState<String, List> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        List<BmsTrdStrategys> bmsTrdStrategies = broadcastState.get("BmsTrdStrategy");
        record.setBmsTrdStrategies(bmsTrdStrategies);
        out.collect(record);
    }

    @Override
    public void processBroadcastElement(Map<String, List> stateRecord, KeyedBroadcastProcessFunction<String, WaringTermalDwmWithStragegyEv, Map<String, List>, WaringTermalDwmWithStragegyEv>.Context ctx, Collector<WaringTermalDwmWithStragegyEv> out) throws Exception {
        BroadcastState<String, List> state = ctx.getBroadcastState(mapStateDescriptor);
        state.clear();
        state.putAll(stateRecord);
    }
}
